package com.orderplus.opshop.synces.core;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
import com.orderplus.opshop.common.util.ValueUtil;
import com.orderplus.opshop.synces.constant.ConfigProperties;
import com.orderplus.opshop.synces.response.CanalResponseData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public abstract class CanalClientFactory {
	@Autowired
	ConfigProperties configProperties;

	public CanalConnector createConnectors() {
		CanalConnector connector = CanalConnectors.newSingleConnector(
				new InetSocketAddress(configProperties.getIp(), configProperties.getPort()), configProperties.getDestination(),
				configProperties.getUsername(), configProperties.getPassword());

		connector.connect();
//		connector.subscribe(".*\\..*");
//		匹配是product开头的表
		connector.subscribe(configProperties.getDbname() + "\\.product.*");
		connector.rollback();

		log.info("start CanalClientFactory");

		return connector;
	}

	public void getData(CanalConnector connector) {
		int batchSize = 1000;

		// 获取指定数量的数据
		Message message = connector.getWithoutAck(batchSize);
		long batchId = message.getId();
		int size = message.getEntries().size();
		if (batchId == -1 || size == 0) {
//			log.info("empty action , sleep 1 millis,wait to next loop");
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
		} else {
			try {
				scanEntry(message.getEntries());
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		// 提交确认
		connector.ack(batchId);
		// 处理失败, 回滚数据
//		connector.rollback(batchId);
	}

	private void scanEntry(List<Entry> entrys) {
		CanalResponseData canalResponseData = new CanalResponseData();

		for (Entry entry : entrys) {
			if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
				continue;
			}

			RowChange rowChage = null;
			try {
				rowChage = RowChange.parseFrom(entry.getStoreValue());
			} catch (Exception e) {
				throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
			}

			EventType eventType = rowChage.getEventType();
			log.info("<================ LogfileName:{},LogfileOffset:{},SchemaName:{},TableName:{},eventType:{}================>",
					entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
					entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType);

			if (ValueUtil.isEmpty(canalResponseData.getTableName())) {
				canalResponseData.setTableName(entry.getHeader().getTableName());
			}

			for (RowData rowData : rowChage.getRowDatasList()) {
				if (eventType == EventType.DELETE) {
//					删除操作
					canalResponseData = this.convertToCanalDeleteResponseData(canalResponseData, rowData);
				} else {
//					新增,修改操作
					canalResponseData = this.convertToCanalResponseData(canalResponseData, rowData);
				}
				canalResponseData.setEventType(eventType);
			}
		}

		if (EventType.INSERT == canalResponseData.getEventType()) {
			addAction(canalResponseData.getTableName(), canalResponseData.getPrimaryKeyName(), canalResponseData.getColumnMap());
		} else if (EventType.UPDATE == canalResponseData.getEventType()) {
			editAction(canalResponseData.getTableName(), canalResponseData.getPrimaryKeyName(), canalResponseData.getColumnMap());
		} else if (EventType.DELETE == canalResponseData.getEventType()) {
			deleteAction(canalResponseData.getTableName(), canalResponseData.getColumnMap());
		}
	}

	public CanalResponseData convertToCanalResponseData(CanalResponseData responseData, RowData rowData) {
		List<Column> columnList = rowData.getAfterColumnsList();
		Map<String, Object> columnMap = responseData.getColumnMap();

		if (ValueUtil.isEmpty(columnMap)) {
			columnMap = new ConcurrentHashMap<>();
		}

		if (ValueUtil.isNotEmpty(columnList)) {
			for (Column column : columnList) {
				if (!column.getIsKey()) {
					if (!column.getUpdated()) {
//						如果不是更新字段,则跳过下一个字段
						continue;
					}
				}

//				判断是否是主键,并且响应主体为空的情况下才进入
				if (ValueUtil.isEmpty(responseData.getPrimaryKeyName()) && column.getIsKey()) {
					responseData.setPrimaryKeyName(column.getName());
				}

				columnMap.put(column.getName(), column.getValue());
			}
		}

		responseData.setColumnMap(columnMap);

		return responseData;
	}

	public CanalResponseData convertToCanalDeleteResponseData(CanalResponseData responseData, RowData rowData) {
		List<Column> columnList = rowData.getBeforeColumnsList();

		Map<String, Object> columnMap = responseData.getColumnMap();

		if (ValueUtil.isEmpty(columnMap)) {
			columnMap = new ConcurrentHashMap<>();
		}

		if (ValueUtil.isNotEmpty(columnList)) {
			for (Column column : columnList) {
//				判断是否是主键,并且响应主体为空的情况下才进入
				if (ValueUtil.isEmpty(responseData.getId()) && column.getIsKey()) {
					responseData.setId(column.getValue());

					columnMap.put(column.getName(), column.getValue());

					break;
				}
			}
		}

		responseData.setColumnMap(columnMap);

		return responseData;
	}

	public abstract void addAction(String tableName, String primaryKeyName, Map<String, Object> map);

	public abstract void editAction(String tableName, String primaryKeyName, Map<String, Object> map);

	public abstract void deleteAction(String tableName, Map<String, Object> map);

}